package com.wsjj.gmall.util;

import com.wsjj.gmall.constant.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.kafka.clients.producer.ProducerConfig;


/*
知识点： 如何保证精确一次性：
    kafkaSource：从kafka主题中读取数据
                通过手动维护便宜量，保证消费的精确一次
    kafkaSink：向kafka主题中写入数据，也可以保证写入的精确一次，需要做如下操作
                开启检查点
                消费端精确一次性配置     生产环境必须写   .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                事务前缀id，必须唯一     生产环境必须写   .setTransactionalIdPrefix(TransactionalIdPrefix)
                设置事务的超时时间：      检查点超时时间 <=  事务的超时时间  <= 事务最大超时时间(默认15分钟)   生产环境必须写     .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*10000+"")
                消费端开启精确一次性：        生产环境必须写    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)    意思是提交在kafka的数据才能读出来，预提交的数据不能读出来，才能保证  精确一次性，开发环境一定要写


 */


////连接kafka的Sink


public class FlinkSinkUtil {
    public static KafkaSink<String> getKafkaSink(String topic, String TransactionalIdPrefix) {


        return KafkaSink.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
//                消费端精确一次性配置    生产环境必须写
//                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//                事务前缀id，必须唯一    生产环境必须写
//                .setTransactionalIdPrefix(TransactionalIdPrefix)
//                设置事务的超时时间：     检查点超时时间 <=  事务的超时时间  <= 事务最大超时时间   生产环境必须写
//                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*10000+"")
                .build();
    }
}
